package com;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class FlinkChineseFilterJob {
    public static void main(String[] args) throws Exception {
        // 获取 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 解析输入参数
        ParameterTool params = ParameterTool.fromArgs(args);
        String inputText = params.get("input", "默认文本123ABC");

        // 创建数据流
        DataStream<String> inputStream = env.fromElements(inputText);

        // 过滤中文字符
        DataStream<String> chineseStream = inputStream.flatMap(new ChineseFilter());

        // 打印输出到 Flink Dashboard
        chineseStream.print();

        // 执行 Flink Job
        env.execute("Flink Chinese Filter Job");
    }

    // 过滤中文的 FlatMapFunction
    public static class ChineseFilter implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) {
            StringBuilder chinese = new StringBuilder();
            for (char c : value.toCharArray()) {
                if (Character.UnicodeScript.of(c) == Character.UnicodeScript.HAN) {
                    chinese.append(c);
                }
            }
            if (chinese.length() > 0) {
                out.collect(chinese.toString());
            }
        }
    }
}

